package com.gjy.pulsar.disruptor;

import com.lmax.disruptor.dsl.Disruptor;
import org.apache.pulsar.client.api.Consumer;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-08-06 12:19:50
 */
public class IntegrationMain {

    public static void main(String[] args) throws Exception {
        // 1. 启动Disruptor
        Disruptor<PulsarMessageEvent> disruptor = DisruptorBootstrapper.startDisruptor();

        // 2. 创建消息桥接器
        PulsarToDisruptorBridge bridge = new PulsarToDisruptorBridge(disruptor.getRingBuffer());

        // 3. 启动Pulsar消费者
        Consumer<byte[]> consumer = PulsarConsumerSetup.initPulsarConsumer(
                "pulsar://localhost:6650",
                "persistent://public/default/orders",
                "disruptor-group",
                bridge
        );
        // 3. 绑定事件处理器
        disruptor.handleEventsWith(new MessageBusinessHandler(consumer));
        disruptor.start();
        System.out.println("系统已启动 >> Pulsar消费 → Disruptor处理");
    }

}
